package org.lql.chaos.kafka.example.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.lql.chaos.kafka.example.pojo.BaseMessage;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.Collections;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 01:19
 * @description:
 **/
@Slf4j
@Service
public class KafkaConsumerExampleService {

	@KafkaListener(
			topics = "topic-string",
			groupId = "group-string",
			containerFactory = "kafkaListenerContainerFactoryString")
	public void listenString(ConsumerRecord<String, Object> record, Acknowledgment ack) {
		int partition = record.partition();
		if (record.value() instanceof BaseMessage) {
			BaseMessage baseMessage = (BaseMessage) record.value();
			log.info("收到消息 topic-string complex: partition={}, key={}, value={}, offset={}", partition, record.key(), baseMessage, record.offset());
		} else {
			log.info("收到消息 topic-string simple: partition={}, key={}, value={}, offset={}", partition, record.key(), record.value(), record.offset());
		}
		// 手动提交 offset
		ack.acknowledge();
	}

	@KafkaListener(topics = "topic-bytes", containerFactory = "kafkaListenerContainerFactoryByte")
	public void listenBytes(ConsumerRecord<String, Object> record) {
		if (record.value() instanceof BaseMessage) {
			BaseMessage baseMessage = (BaseMessage) record.value();
			log.info("收到消息 topic-bytes complex: key={}, value={}, offset={}", record.key(), baseMessage, record.offset());
		} else {
			log.info("收到消息 topic-bytes simple: key={}, value={}, offset={}", record.key(), record.value(), record.offset());
		}
	}

	@KafkaListener(topics = "topic-json", containerFactory = "kafkaListenerContainerFactoryJson")
	public void listenJson(ConsumerRecord<String, Object> record) {
		if (record.value() instanceof BaseMessage) {
			BaseMessage baseMessage = (BaseMessage) record.value();
			log.info("收到消息 topic-json complex: key={}, value={}, offset={}", record.key(), baseMessage, record.offset());
		} else {
			log.info("收到消息 topic-json simple: key={}, value={}, offset={}", record.key(), record.value(), record.offset());
		}
	}

	@KafkaListener(topics = "topic-kryo", containerFactory = "kafkaListenerContainerFactoryKryo")
	public void listenKryo(ConsumerRecord<String, Object> record) {
		if (record.value() instanceof BaseMessage) {
			BaseMessage baseMessage = (BaseMessage) record.value();
			log.info("收到消息 topic-kryo complex: key={}, value={}, offset={}", record.key(), baseMessage, record.offset());
		} else {
			log.info("收到消息 topic-kryo simple: key={}, value={}, offset={}", record.key(), record.value(), record.offset());
		}
	}

	@KafkaListener(topics = "topic-string", containerFactory = "kafkaListenerContainerFactoryString")
	public void listenString(ConsumerRecord<String, Object> record, Consumer<?, ?> consumer) {
		int partition = record.partition();
		log.info("收到消息: partition={}, key={}, value={}, offset={}",
				partition, record.key(), record.value(), record.offset());

		// 异步提交当前 offset + 1
		TopicPartition topicPartition = new TopicPartition(record.topic(), partition);
		OffsetAndMetadata offsetMeta = new OffsetAndMetadata(record.offset() + 1);
		consumer.commitAsync(Collections.singletonMap(topicPartition, offsetMeta), (offsets, ex) -> {
			if (ex == null) {
				log.info("异步提交成功: {}", offsets);
			} else {
				log.error("异步提交失败: {}", ex.getMessage(), ex);
			}
		});
	}

}
